package raft

import (
	"bufio"
	"distributed_db/master"
	"distributed_db/utils"
	"encoding/gob"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// Here provide a standard template to create a raft server

type Server struct {
	mu           sync.Mutex
	me           int
	rf           *Raft         // raft service
	applyCh      chan ApplyMsg // apply channel, where you can get the committed command you want start in Start()
	listener     net.Listener  // tcp listener
	dead         int32         // set by Kill()
	dying        *sync.Cond
	master       *master.Master // master service
	clientRecord map[int64]*ClientCommit
	Index        int      // committed index// debug
	Log          []LogMsg // debug
}

type ClientCommit struct {
	record uint32
	cond   *sync.Cond
}

const (
	Table        = 0
	RegionServer = 1
	Empty        = 2
)
const (
	Create = 0
	Delete = 1
	Update = 2
)

// 一致性统一日志信息
type LogMsg struct {
	Type        int
	Info        interface{}
	Op          int
	CommitIndex int
	SerialNum   uint32
	ClientId    int64
}

// debug: 测试用
func (s *Server) GetMaster() *master.Master {
	return s.master
}

func (s *Server) get_client_pair(key int64) (uint32, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	v, ok := s.clientRecord[key]
	if !ok {
		return 0, ok
	}
	return v.record, ok
}

func (s *Server) set_client_pair(key int64, record uint32) {
	s.mu.Lock()
	if _, exist := s.clientRecord[key]; !exist {
		s.clientRecord[key] = &ClientCommit{
			record: record,
			cond:   sync.NewCond(&sync.Mutex{}),
		}
	} else {
		s.clientRecord[key].record = record
	}
	// raft.LogC(raft.Yellow, "S%d Notify %d\n", s.me, record)
	s.mu.Unlock()
	s.NotifyClientValue(key)
}

func (s *Server) WaitClientValue(client int64, value uint32) bool {
	s.clientRecord[client].cond.L.Lock()
	defer s.clientRecord[client].cond.L.Unlock()

	// 设置超时时间
	timer := time.After(time.Second)

	utils.LogC(utils.Yellow, "S%d:%d Wait client %d\n", s.me, s.clientRecord[client].record, value)

	for value > s.clientRecord[client].record {
		// 使用 select 实现带超时的等待
		select {
		case <-timer:
			// 超时处理逻辑
			return false
		default:
			// 等待通知
			s.clientRecord[client].cond.Wait()
		}
	}
	utils.LogC(utils.Yellow, "S%d:%d Wait client %d done\n", s.me, s.clientRecord[client].record, value)
	return true
}

func (s *Server) NotifyClientValue(client int64) {
	s.clientRecord[client].cond.Broadcast()
}

func (s *Server) get_commit_index() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.Index
}

func (s *Server) set_commit_index(index int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.Index = index
}

func (s *Server) WaitCommitValue(index int) bool {
	// 设置超时时间
	timer := time.After(time.Second)

	// r.LogC(utils.Yellow, "S%d:%d Wait commit %d\n", s.me, s.get_commit_index(), index)

	for index > s.get_commit_index() {
		select {
		case <-timer:
			// 超时处理逻辑
			return false
		default:
			// 等待通知
			time.Sleep(time.Millisecond)
		}
	}
	return true
}

func (s *Server) ApplyLogMsg(t int, info interface{}, op int) (string, bool) {
	switch t {
	case Table:
		switch op {
		case Create:
			args := info.(utils.CreateTableArgs)
			cols := make([]string, 0)
			pk := make([]string, 0)
			index := make([]string, 0)
			for colName := range args.Cols {
				cols = append(cols, colName)
			}
			index = append(index, "test_index:"+cols[0])
			pk = append(pk, cols[0])
			s.master.NewTable(args.TableName, cols, pk, index)
		case Delete:
			args := info.(utils.DropTableArgs)
			s.master.DeleteTable(args.TableName)
		case Update:
			// TODO: update table
		default:
		}
	case RegionServer:
	case Empty:
	default:
	}
	return "", false
}

func (s *Server) KillServer() {
	s.rf.Kill()
	s.listener.Close()
	s.master.Shutdown()
	utils.LogC(utils.Purple, "Server %d killed with %d\n", s.me, s.rf.commitIndex)
	atomic.StoreInt32(&s.dead, 1)
	s.dying.Broadcast()
}

func (s *Server) server_killed() bool {
	z := atomic.LoadInt32(&s.dead)
	return z == 1
}

// StartRaftService when you call this function, the server will start to provide service as a raft server
func (s *Server) StartRaftService() {
	s.rf.StartRaftService()
}

func (s *Server) GetState() (int, bool) {
	return s.rf.GetState()
}

// MakePeers make client End from config file
func MakePeers(filename string) []*utils.ClientEnd {
	file, err := os.Open(filename)
	if err != nil {
		return nil
	}
	defer func() {
		if closeErr := file.Close(); closeErr != nil {
			log.Printf("Failed to close file: %v\n", closeErr)
		}
	}()

	clients := make([]*utils.ClientEnd, 0)
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		serverEnd := strings.TrimSpace(scanner.Text())
		if serverEnd != "" {
			client := utils.MakeEnd(serverEnd)
			clients = append(clients, client)
		}
	}

	if err := scanner.Err(); err != nil {
		log.Fatalf("Scanner Err: %v", err)
	}
	return clients
}

// UpdateCache 测试阶段：更新缓存
func (s *Server) UpdateCache(args *utils.UpdateCacheArgs, reply *utils.UpdateCacheReply) error {
	if s.master == nil {
		reply.Err = utils.ErrMasterNotReady
		return fmt.Errorf("master not ready")
	}
	leader := s.rf.GetLeaderId()
	if leader != s.me {
		reply.Err = utils.ErrWrongLeader
		return nil
	}
	// check whether current entry id has been served
	v, ok := s.get_client_pair(args.ClientId)
	if !ok {
		s.set_client_pair(args.ClientId, args.EntryId-1)
		v = args.EntryId - 1
	}
	// if not serverd, wait for previous
	if v < args.EntryId {
		if !s.WaitClientValue(args.ClientId, args.EntryId-1) {
			reply.Err = ""
			return fmt.Errorf(utils.ErrTimeWait)
		}
		s.set_client_pair(args.ClientId, args.EntryId)
	}
	reply.RegionServers = make(map[string]map[string]int)
	utils.LogC(utils.Green, "UpdateCacheArgs: %v\n", args)
	for _, tableName := range args.TableNames {
		table, ok := s.master.GetTables()[tableName]
		if !ok {
			reply.Err = utils.ErrNoSuchTableInMemory
			return fmt.Errorf("no such table in memory: %s", tableName)
		}
		reply.RegionServers[tableName] = make(map[string]int, 0)
		for _, region := range table.Regions {
			for _, rs := range s.master.GetRegionServers() {
				if rs.ID == region.RegionServerID {

					reply.RegionServers[tableName][rs.IP+":"+rs.Port] = region.RecordNum
					break
				}
			}
		}
	}
	reply.TTL = 20
	reply.Err = utils.OK
	empty_log := LogMsg{
		Type:      Empty,
		Info:      "",
		Op:        0,
		ClientId:  args.ClientId,
		SerialNum: args.EntryId,
	}
	s.rf.Start(empty_log)
	return nil
}

func (s *Server) CreateTable(args *utils.CreateTableArgs, reply *utils.CreateTableReply) error {
	for {
		if s.master == nil {
			reply.Err = utils.ErrMasterNotReady
			return fmt.Errorf("master not ready")
		}
		leader := s.rf.GetLeaderId()
		if leader != s.me {
			reply.Err = utils.ErrWrongLeader
			return nil
		}
		// check whether current entry id has been served
		v, ok := s.get_client_pair(args.ClientId)
		if !ok {
			s.set_client_pair(args.ClientId, args.EntryId-1)
			v = args.EntryId - 1
		}
		// if serverd, return
		if v >= args.EntryId {
			reply.Err = utils.OK
			return nil
		}
		log := LogMsg{
			Type:      Table,
			Info:      *args,
			Op:        Create,
			ClientId:  args.ClientId,
			SerialNum: args.EntryId,
		}
		// 一致性处理
		idx, _, ok := s.rf.Start(log)
		if !ok {
			reply.Err = utils.ErrWrongLeader
			return nil
		}
		judge := s.WaitCommitValue(idx)
		if !judge {
			reply.Err = utils.ErrTimeWait
			return fmt.Errorf(utils.ErrTimeWait)
		}
		v, _ = s.get_client_pair(args.ClientId)
		if v == args.EntryId {
			// if committed as the current client and entry
			reply.Err = utils.OK
			// cols := make([]string, 0)
			// for colName := range args.Cols {
			// 	cols = append(cols, colName)
			// }
			// pk := make([]string, 0)
			// index := make([]string, 0)
			// pk = append(pk, cols[0])
			// index = append(index, cols[0])
			// s.master.NewTable(args.TableName, cols, pk, index)

			// reply.RegionServers = make([]string, 0)
			// for _, rs := range s.master.GetTables()[args.TableName].Regions {
			// 	reply.RegionServers = append(reply.RegionServers, rs.RegionName)
			// }
			return nil
		}
	}
}

func (s *Server) DropTable(args *utils.DropTableArgs, reply *utils.DropTableReply) error {
	for {
		if s.master == nil {
			reply.Err = utils.ErrMasterNotReady
			return fmt.Errorf("master not ready")
		}
		leader := s.rf.GetLeaderId()
		if leader != s.me {
			reply.Err = utils.ErrWrongLeader
			return nil
		}
		// check whether current entry id has been served
		v, ok := s.get_client_pair(args.ClientId)
		if !ok {
			s.set_client_pair(args.ClientId, args.EntryId-1)
			v = args.EntryId - 1
		}
		// if serverd, return
		if v >= args.EntryId {
			reply.Err = utils.OK
			return nil
		}

		log := LogMsg{
			Type:      Table,
			Info:      *args,
			Op:        Delete,
			ClientId:  args.ClientId,
			SerialNum: args.EntryId,
		}
		// 一致性处理
		idx, _, ok := s.rf.Start(log)
		if !ok {
			reply.Err = utils.ErrWrongLeader
			return nil
		}
		judge := s.WaitCommitValue(idx)
		if !judge {
			reply.Err = utils.ErrTimeWait
			return fmt.Errorf(utils.ErrTimeWait)
		}
		v, _ = s.get_client_pair(args.ClientId)
		if v == args.EntryId {
			// if committed as the current client and entry
			reply.Err = utils.OK
			return nil
		}
	}
}

func (s *Server) InsertData(args *utils.InsertDataArgs, reply *utils.InsertDataReply) error {
	if s.master == nil {
		reply.Err = utils.ErrMasterNotReady
		return fmt.Errorf("master not ready")
	}
	leader := s.rf.GetLeaderId()
	if leader != s.me {
		reply.Err = utils.ErrWrongLeader
		return nil
	}
	// check whether current entry id has been served
	v, ok := s.get_client_pair(args.ClientId)
	if !ok {
		s.set_client_pair(args.ClientId, args.EntryId-1)
		v = args.EntryId - 1
	}
	// if not serverd, wait for previous
	if v < args.EntryId {
		utils.LogC(utils.Green, "stage6:clientID:%d,entryID:%d\n", args.ClientId, args.EntryId)
		if !s.WaitClientValue(args.ClientId, args.EntryId-1) {
			reply.Err = ""
			utils.LogC(utils.Red, "%dWaitClientValue failed\n", s.me)
			return nil
		}
		s.set_client_pair(args.ClientId, args.EntryId)
	}
	cols := make([]string, 0)
	data := make([][]interface{}, 0)
	data = append(data, make([]interface{}, 0))
	for colName, colValue := range args.Data {
		cols = append(cols, colName)
		data[0] = append(data[0], colValue)

	}
	rows, err := s.master.InsertDataToTable("", args.TableName, cols, data)
	if err != nil {
		reply.Err = utils.ErrInsertData
		utils.LogC(utils.Red, "InsertDataToTable failed: %v\n", err)
		return err
	}
	reply.Rows = int64(rows)
	reply.Err = utils.OK
	empty_log := LogMsg{
		Type:      Empty,
		Info:      "",
		Op:        0,
		ClientId:  args.ClientId,
		SerialNum: args.EntryId,
	}
	s.rf.Start(empty_log)
	return nil
}

func (s *Server) DeleteData(args *utils.DeleteDataArgs, reply *utils.DeleteDataReply) error {
	if s.master == nil {
		reply.Err = utils.ErrMasterNotReady
		return fmt.Errorf("master not ready")
	}
	leader := s.rf.GetLeaderId()
	if leader != s.me {
		reply.Err = utils.ErrWrongLeader
		return nil
	}
	// check whether current entry id has been served
	v, ok := s.get_client_pair(args.ClientId)
	if !ok {
		s.set_client_pair(args.ClientId, args.EntryId-1)
		v = args.EntryId - 1
	}
	// if not serverd, wait for previous
	if v < args.EntryId {
		if !s.WaitClientValue(args.ClientId, args.EntryId-1) {
			reply.Err = ""
			return nil
		}
		s.set_client_pair(args.ClientId, args.EntryId)
	}
	rows, err := s.master.DeleteDataFromTable("", args.TableName, args.Key)
	if err != nil {
		reply.Err = utils.ErrDeleteData
		utils.LogC(utils.Red, "DeleteDataFromTable failed: %v\n", err)
		return err
	}
	reply.Rows = int64(rows)
	reply.Err = utils.OK
	empty_log := LogMsg{
		Type:      Empty,
		Info:      "",
		Op:        0,
		ClientId:  args.ClientId,
		SerialNum: args.EntryId,
	}
	s.rf.Start(empty_log)
	return nil
}

// MakeServer make server by all other servers' client end as each server havs to communicate with each other
// define port number and me as the server's id
// recommend to set me as the sequence number of the server in config file
func MakeServer(servers []*utils.ClientEnd, port string, me int) *Server {
	gob.Register(utils.CreateTableArgs{})
	gob.Register(utils.DropTableArgs{})
	gob.Register(LogMsg{})
	server := &Server{}
	server.applyCh = make(chan ApplyMsg)
	server.me = me
	server.rf = Make(servers, me, MakePersister(fmt.Sprint(me)), server.applyCh)
	server.dying = sync.NewCond(&sync.Mutex{})
	server.master = master.MaterInit(port)
	server.clientRecord = make(map[int64]*ClientCommit)
	server.Index = 0
	server.Log = make([]LogMsg, 0)

	s := rpc.NewServer()
	s.Register(server.rf)
	s.Register(server)

	listener, err := net.Listen("tcp", ":"+port)
	if err != nil {
		log.Fatal("ListenTCP error:", err)
		return nil
	}
	fmt.Printf("listening %s\n", port)
	server.listener = listener

	// a new route to handle the connection
	go http.Serve(listener, s)

	// a goroutine to handle the apply channel
	go func() {
		for m := range server.applyCh {
			if !m.CommandValid {
				// ignore other types of ApplyMsg
			} else {
				// inc committed log

				v := m.Command
				op, ok := v.(LogMsg)
				if !ok {
					utils.LogC(utils.Red, "Type assertion failed\n")
					continue
				}
				if len(server.Log) == m.CommandIndex-1 {
					server.Log = append(server.Log, op)
				}
				// get client entry size
				entry, ok := server.get_client_pair(op.ClientId)
				if !ok { // no such client
					server.set_client_pair(op.ClientId, op.SerialNum-1)
					entry = op.SerialNum - 1
				}
				// if op num is greater then current entry size
				if entry < op.SerialNum {
					// apply op
					server.ApplyLogMsg(op.Type, op.Info, op.Op)
					// update client serial number
					entry = op.SerialNum
					server.set_client_pair(op.ClientId, entry)
					// raft.LogC(raft.Blue, "%s at S%d K: %v, V: %v\n", op.Operation, server.me, op.Key, op.Value)
				}
				if m.CommandIndex > server.get_commit_index() {
					server.set_commit_index(m.CommandIndex)
				}
			}
		}
	}()

	go func() {
		server.dying.L.Lock()
		for !server.server_killed() {
			server.dying.Wait()
		}
		server.dying.L.Unlock()

		close(server.applyCh)
	}()
	return server
}
